/*
 * 文件名：CommonUnderTask.java
 * 版权：Copyright 2013-2013 Hugedata Tech. Co. Ltd. All Rights Reserved
 * 描述：〈描述〉
 * 修改人：〈修改人〉
 * 修改时间：YYYY-MM-DD
 * 跟踪单号：〈跟踪单号〉
 * 修改单号：〈修改单号〉
 * 修改内容：〈修改内容〉
 */
package com.hugedata.cdnserver.common;

import java.io.BufferedReader;
import java.io.BufferedWriter;
import java.io.File;
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.FileReader;
import java.io.IOException;
import java.io.OutputStreamWriter;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Properties;

import org.apache.commons.lang.StringUtils;
import org.apache.log4j.Logger;
import org.apache.zookeeper.common.IOUtils;

import com.hugedata.cdn.util.GlobalConstantDef;
import com.hugedata.cdnserver.util.SystemUtils;
import com.hugedata.imsp.jabberc.IMSPClient;
import com.hugedata.imsp.jabberc.service.node.IDCNode;
import com.hugedata.imsp.jabberc.service.node.IDCTaskNode;
import com.hugedata.imsp.jabberc.service.node.NodeOnConditionRequest;
import com.hugedata.imsp.jabberc.service.node.NodeOnConditionResponse;
import com.hugedata.imsp.jabberc.service.task.TaskOnConditionRequest;
import com.hugedata.imsp.jabberc.service.task.TaskOnConditionResponse;

/**
 * 
 * @ClassName: CommonUnderTask
 * @Description: 下发任务
 * @author LuXiuMin
 * @date 2013-8-13 上午9:10:20
 * 
 */
public final class CommonUnderTask
{
    private final static Logger LOGGER = Logger.getLogger(CommonUnderTask.class);

    public static void main(String[] args)
    {
        // hadoop.tmp.dir
    }

    /**
     * constructor
     */
    private CommonUnderTask()
    {

    }

    public static List<IDCNode> getAvailableNode(IMSPClient client, String taskName, String toolType)
            throws IOException
    {
        // 获取可用使用的节点数
        LOGGER.info(taskName + ",发送获取可用使用的节点数请求,请求类型是：查询全部节点.");
        NodeOnConditionRequest request = new NodeOnConditionRequest();
        request.setToolId(SystemUtils.getCfg(GlobalConstantDef.CONFIG_PATH).getProperty(toolType));
        request.setQueryType(0);

        NodeOnConditionResponse res = null;
        try
        {
            res = client.getNodesByToolAndISP(request);
        }
        catch (Exception e1)
        {
            throw new RuntimeException("NodeResponse getNodes Exception", e1);
        }

        List<IDCNode> nodeList = res.getIdcList();
        try
        {
            String repeatNumber = SystemUtils.getCfg(GlobalConstantDef.CONFIG_PATH).getProperty(
                    "get_availablenode_task_send_fail_repeat_number");
            int repeatNum = Integer.valueOf(repeatNumber);
            String sleepTime = SystemUtils.getCfg(GlobalConstantDef.CONFIG_PATH).getProperty(
                    "get_availablenode_task_send_fail_repeat_time");

            int number = 0;
            while (nodeList == null || nodeList.size() == 0)
            {
                number++;
                if (number > repeatNum)
                {
                    break;
                }
                LOGGER.info(taskName + ",获取节点失败,暂停" + (Integer.valueOf(sleepTime)) + "秒后,会重复获取节点...");
                LOGGER.info(taskName + ",第" + number + "次重复下发...");
                Thread.sleep(Integer.valueOf(sleepTime) * 1000);
                nodeList = res.getIdcList();
            }
        }
        catch (Exception e)
        {
            LOGGER.error(taskName + ",转换任务发送失败重复间隔发送次数为整型错误" + e);
            return null;
        }
        if (nodeList == null)
        {
            LOGGER.error(taskName + ",获取节点失败");
            return null;
        }
        else
        {
            if (nodeList.size() == 0)
            {
                LOGGER.error(taskName + ",获取节点失败");
                return null;
            }
        }
        LOGGER.info(taskName + ",可用节点数:" + nodeList.size());
        return nodeList;
    }

    public static int createTask(IMSPClient client, String taskName, String toolType, File file, IDCNode idcnode,
            String beginDateOfCfg) throws IOException
    {
        LOGGER.info("createTask method in....");
        Properties cfg = SystemUtils.getCfg(GlobalConstantDef.CONFIG_PATH);

        String repeatNumber = "";
        String sleepTime = "";
        String toolTypeStr = cfg.getProperty(toolType).trim();

        String crawlToolRef = cfg.getProperty("crawl_tool_type_reference").trim();
        String dnsParseToolRef = cfg.getProperty("dns_parse_tool_type_reference").trim();
        String dnsDelayToolRef = cfg.getProperty("dns_delay_tool_type_reference").trim();
        String qosToolRef = cfg.getProperty("qos_tool_type_reference").trim();
        String tracerouteToolRef = cfg.getProperty("traceroute_tool_type_reference").trim();

        String taskBeginDate = SystemUtils.getTaskBeginTime(beginDateOfCfg);
        String taskEndDate = SystemUtils.getTaskEndDate(new Date());

        int toolRunInterval = -1;
        int taskForceendFlag = -1;

        // 设置优先级
        int taskpriority = 3;
        // 爬虫工具
        if (toolTypeStr.equals(crawlToolRef))
        {
            taskEndDate = SystemUtils.getTaskEndDateOfCrawleTool();
            toolRunInterval = Integer.valueOf(SystemUtils.getCfg(GlobalConstantDef.CONFIG_PATH).getProperty(
                    "crawler_run_interval"));
            taskForceendFlag = 0;
            LOGGER.info("爬虫工具,任务开始时间:" + taskBeginDate + " 任务结束时间:" + taskEndDate + " 工具每" + toolRunInterval + "分钟运行一次");
        }// dns解析工具
        else if (toolTypeStr.equals(dnsParseToolRef))
        {
            toolRunInterval = Integer.valueOf(cfg.getProperty("dns_parse_run_interval"));

            String forceStringForDnsParse = cfg.getProperty("dnsparse_tool_task_foceendflag");
            taskForceendFlag = Integer.valueOf(forceStringForDnsParse).intValue();
            String string = "";
            if (taskForceendFlag == 0)
            {
                string = "任务优先";
            }
            else if (taskForceendFlag == 1)
            {
                string = "时间优先";
            }

            LOGGER.info("dns解析工具,任务开始时间:" + taskBeginDate + " 任务结束时间:" + taskEndDate + " 工具每" + toolRunInterval
                    + "分钟运行一次" + " 任务强行结束标志(任务优先级别)是:" + taskForceendFlag + " 也即:" + string);
        }// qos工具
        else if (toolTypeStr.equals(qosToolRef))
        {
            toolRunInterval = Integer.valueOf(cfg.getProperty("qos_run_interval"));

            String qosToolForceenFlag = cfg.getProperty("qos_tool_task_foceendflag").trim();

            taskForceendFlag = Integer.valueOf(qosToolForceenFlag);
            String string = "";
            if (taskForceendFlag == 0)
            {
                string = "任务优先";
            }
            else if (taskForceendFlag == 1)
            {
                string = "时间优先";
            }
            LOGGER.info("qos工具,任务开始时间:" + taskBeginDate + " 任务结束时间:" + taskEndDate + " 工具每" + toolRunInterval
                    + "分钟运行一次" + " 任务强行结束标志(任务优先级别)是:" + taskForceendFlag + " 也即:" + string);

        }// DNS延时工具
        else if (toolTypeStr.equals(dnsDelayToolRef))
        {
            toolRunInterval = Integer.valueOf(cfg.getProperty("dns_delay_run_interval"));

            String dnsDelayToolForceenFlag = cfg.getProperty("dnsdelay_tool_task_foceendflag").trim();
            taskForceendFlag = Integer.valueOf(dnsDelayToolForceenFlag);
            String string = "";
            if (taskForceendFlag == 0)
            {
                string = "任务优先";
            }
            else if (taskForceendFlag == 1)
            {
                string = "时间优先";
            }
            LOGGER.info("DNS延时工具,任务开始时间:" + taskBeginDate + " 任务结束时间:" + taskEndDate + " 工具每" + toolRunInterval
                    + "分钟运行一次" + " 任务强行结束标志(任务优先级别)是:" + taskForceendFlag + " 也即:" + string);
        }//traceroute延时工具
        else if (toolTypeStr.equals(tracerouteToolRef))
        {
            toolRunInterval = Integer.valueOf(cfg.getProperty("traceroute_run_interval"));
            
            String tracerouteToolForceenFlag = cfg.getProperty("traceroute_tool_task_foceendflag").trim();
            taskForceendFlag = Integer.valueOf(tracerouteToolForceenFlag);
            String string = "";
            if (taskForceendFlag == 0)
            {
                string = "任务优先";
            }
            else if (taskForceendFlag == 1)
            {
                string = "时间优先";
            }
            LOGGER.info("traceroute延时工具,任务开始时间:" + taskBeginDate + " 任务结束时间:" + taskEndDate + " 工具每" + toolRunInterval
                    + "分钟运行一次" + " 任务强行结束标志(任务优先级别)是:" + taskForceendFlag + " 也即:" + string);
        }

        int resultCode = -1;
        int taskUnderType = Integer.valueOf(cfg.getProperty("task_under_type"));
        String underFileUrl = cfg.getProperty("postUrl");
        TaskOnConditionRequest request = new TaskOnConditionRequest();

        request.setToolId(toolTypeStr);
        request.setBeginTime(taskBeginDate);
        request.setEndTime(taskEndDate);
        request.setFileSize((int) file.length());
        request.setForceEndFlag(taskForceendFlag);
        request.setInputType(taskUnderType);
        request.setInput(file.getAbsolutePath());

        // 这些数据需要前台传入
        List<IDCTaskNode> idcList = new ArrayList<IDCTaskNode>();
        IDCTaskNode node = new IDCTaskNode();
        node.setAreaCode(idcnode.getAreaCode());
        node.setMnCount(idcnode.getMnCount());
        node.setIspCode(idcnode.getIspCode());
        idcList.add(node);
        request.setIdcList(idcList);

        request.setOutput(underFileUrl);
        request.setOutputType(1);
        request.setTaskInterval(toolRunInterval);
        request.setImmediate(0);

        // 设置优先级
        request.setTaskPriority(taskpriority);

        TaskOnConditionResponse response;
        try
        {
            response = client.createTaskByArea(request);
        }
        catch (Exception e1)
        {
            throw new RuntimeException("TaskResponse createTask exception", e1);
        }
        resultCode = response.getResultcode();
        LOGGER.info(taskName + ",结果码:" + resultCode + " mcs返回信息:" + response.getDesc() + " 任务ID:"
                + response.getTaskId());

        // 爬虫和dns解析工具
        if (toolTypeStr.equals(crawlToolRef) || toolTypeStr.equals(dnsParseToolRef))
        {
            repeatNumber = cfg.getProperty("task_send_fail_repeat_number");
            sleepTime = cfg.getProperty("task_send_fail_repeat_time");
        }// qos任务和DNS延时工具
        else if (toolTypeStr.equals(dnsDelayToolRef) || toolTypeStr.equals(qosToolRef))
        {
            repeatNumber = cfg.getProperty("qos_dns_task_send_fail_repeat_number");
            sleepTime = cfg.getProperty("qos_dns_task_send_fail_repeat_time");
        }//traceroute工具
        else if (toolTypeStr.equals(tracerouteToolRef))
        {
            repeatNumber = cfg.getProperty("qos_dns_task_send_fail_repeat_number");
            sleepTime = cfg.getProperty("qos_dns_task_send_fail_repeat_time");
        }
        int repeatNum = Integer.valueOf(repeatNumber);
        int number = 0;
        while (resultCode != 0)
        {
            number++;
            if (number > repeatNum)
            {
                break;
            }
            LOGGER.info(taskName + "失败,暂停" + (Integer.valueOf(sleepTime)) + "秒后,会重复下发任务...");
            LOGGER.info(taskName + ",第" + number + "次重复下发...");

            try
            {
                Thread.sleep(Integer.valueOf(sleepTime) * 1000);
            }
            catch (NumberFormatException e)
            {
                LOGGER.error("暂停" + taskName + "数字转换失败" + e);
                resultCode = -1;
                return resultCode;
            }
            catch (InterruptedException e)
            {
                LOGGER.error("暂停" + taskName + "睡眠失败 " + e);
                resultCode = -1;
                return resultCode;
            }
            try
            {
                response = client.createTaskByArea(request);
            }
            catch (Exception e)
            {
                throw new RuntimeException("TaskResponse createTask exception", e);
            }
        }
        LOGGER.info("createTask method out....");
        return resultCode;
    }

    public static File getTop100UnderFile(boolean repeatUnderFlag, String taskName, File underFile, int splitFileLine,
            int nodeSize, int loopNum, int underOkFileNumber, BufferedReader br) throws IOException
    {
        File file = null;
        if (repeatUnderFlag == false)
        {
            file = splitFile(splitFileLine, nodeSize, loopNum, br, underFile);
            return file;
        }
        else if (repeatUnderFlag == true)
        {
            // 文件下发成功数等于0,则重新从头开始读Top文件
            if (underOkFileNumber == 0)
            {
                IOUtils.closeStream(br);
                br = readTopFile();
            }
            file = splitFile(splitFileLine, nodeSize, loopNum, br, underFile);
        }
        return file;
    }

    // 处理爬虫任务下发
    public static int handleCrawleTaskUnder(IMSPClient client, List<IDCNode> nodeList, String taskName,
            String taskBeginTime, File underFile) throws IOException
    {
        int resultcode = -1;
        int nodeSize = nodeList.size();

        // 获取下发爬虫任务失败,重试次数
        String repeatNumStr = SystemUtils.getCfg(GlobalConstantDef.CONFIG_PATH).getProperty("task_send_fail_repeat_time");
        int repeatNum = Integer.valueOf(repeatNumStr);

        // 获取Top文件行数
        int seedsFileLine = SystemUtils.getSeedsFileLineNumbers(GlobalConstantDef.CONFIG_PATH);
        // 拆分的小文件行数
        int splitFileLine = seedsFileLine / nodeSize;

        int offsetNum = 0;
        // 不能整除
        if (seedsFileLine % nodeSize != 0)
        {
            int tempTotalFileLineNum = splitFileLine * nodeSize;
            offsetNum = seedsFileLine - tempTotalFileLineNum;
            // 要在前这些的节点上每个增加一行,以平衡节点任务
            splitFileLine = splitFileLine + 1;
        }

        // 拆分的小文件个数
        int fileNum = nodeSize;
        // 下发成功文件个数
        int underOkFileNumber = 0;
        // 下发失败文件个数
        int fileUnderFailNumber = 0;
        // 是否重试过标志
        boolean repeatUnderFlag = false;

        List<NodeUseInfo> useList = new ArrayList<NodeUseInfo>();
        // FileOutputStream fileOutStream = null;
        // OutputStreamWriter outSW = null;
        // BufferedWriter bw = null;
        BufferedReader br = null;
        try
        {
            br = readTopFile();
            int newSplitFileLine = splitFileLine;

            // 循环所有可用节点,下发所有任务
            One: for (int i = 0; i < nodeSize; i++)
            {
                IDCNode nodeInfo = nodeList.get(i);
                // String mnidStr = nodeInfo.getMnId();
                // int mnId = Integer.valueOf(mnidStr);

                // 不能整除
                if (seedsFileLine % nodeSize != 0)
                {
                    if (i >= offsetNum)
                    {
                        newSplitFileLine = splitFileLine - 1;
                    }
                }

                // 获取下发文件
                File file = getTop100UnderFile(repeatUnderFlag, taskName, underFile, newSplitFileLine, nodeSize, i,
                        underOkFileNumber, br);
                if (file == null || file.length() == 0)
                {
                    LOGGER.info("seedtop文件不存在或者没有内容");
                    break;
                }

                // 下发任务
                // 新pd修改下发任务传入idcnode
                resultcode = CommonUnderTask.createTask(client, taskName, "crawl_tool_type", file, nodeInfo,
                        taskBeginTime);

                // 下发成功
                if (resultcode == 0)
                {
                    underOkFileNumber++;
                    NodeUseInfo useNode = new NodeUseInfo();
                    useNode.setAvailableFlag(true);
                    // useNode.setMnCode(mnId);
                    useNode.setNode(nodeInfo);

                    useNode.setUseFlag(true);
                    useList.add(useNode);
                    file.delete();
                    continue;
                } // 下发失败,试着向下一个节点下发文件
                else
                {
                    while (resultcode != 0)
                    {
                        // 所有节点都试过,并且失败
                        if (i == (nodeList.size() - 1))
                        {
                            break;
                        }
                        // 使用同一个文件,获取下一个节点,试下发任务
                        IDCNode temp = nodeList.get(++i);
                        // String newMnid = temp.getMnId();
                        // Integer newIntMnid = Integer.valueOf(newMnid);
                        // 下发任务
                        resultcode = CommonUnderTask.createTask(client, taskName, "crawl_tool_type", file, temp,
                                taskBeginTime);
                        // 下发成功
                        if (resultcode == 0)
                        {
                            underOkFileNumber++;
                            NodeUseInfo useNode = new NodeUseInfo();
                            useNode.setAvailableFlag(true);
                            // useNode.setMnCode(newIntMnid.intValue());
                            useNode.setNode(nodeInfo);
                            useNode.setUseFlag(true);
                            useList.add(useNode);
                            // 删除临时文件
                            file.delete();
                            break;
                        }
                    }
                }

                // 所有节点都已试下发过,
                if (i == (nodeList.size() - 1))
                {
                    fileUnderFailNumber++;
                    // 但还有未下发过的文件(余下一些文件)
                    if (underOkFileNumber < fileNum)
                    {
                        // 试下发任务超出配置次数,则不再试下发
                        if (fileUnderFailNumber >= repeatNum)
                        {
                            break One;
                        }
                        // 那么,第二次再获取可用节点,重新试下发
                        List<IDCNode> nodeListNew = CommonUnderTask.getAvailableNode(client, taskName,
                                "crawl_tool_type");
                        // 第二次获取的可用节点数大于上一次可用节点数,使用新的节点数去下发文件
                        if (nodeListNew.size() > useList.size())
                        {
                            nodeList = nodeListNew;
                            i = -1;
                            repeatUnderFlag = true;
                        }
                        else
                        // 第二次获取的可用节点数小于上一次可用节点数,使用上一次的节点数去下发文件
                        {
                            for (NodeUseInfo nodeUseInfo : useList)
                            {
                                // NodeInfo nIF = new NodeInfo();
                                // nIF.setMnCode(String.valueOf(nodeUseInfo
                                // .getMnCode()));
                                // nIF.setMnId(String.valueOf(nodeUseInfo.getMnCode()));
                                nodeList.add(nodeUseInfo.getNode());
                            }
                            i = -1;
                            repeatUnderFlag = true;
                        }
                    }
                }
            }
        }
        catch (FileNotFoundException e)
        {
            throw new RuntimeException(e);
        }
        finally
        {
            IOUtils.closeStream(br);
        }
        if (underOkFileNumber == 0)
        {
            LOGGER.info("所有节点都多次试验,结果没有一个文件下发成功,因此,所有的节点都不可用");
        }
        return resultcode;
    }

    private static File splitFile(int splitFileLine, int nodeSize, int loopNum, BufferedReader br, File seedsFile)
    {
        String codeString = "";
        String seedTopFileLineSeparator = "";
        String newLine = "";
        try
        {
            String seedTopFileName = SystemUtils.getCfg(GlobalConstantDef.CONFIG_PATH).getProperty("top100_file_name");
            LOGGER.info("top100_file_name " + seedTopFileName + " splitFileLine " + splitFileLine);

            codeString = SystemUtils.getFileEncode(seedTopFileName);
            LOGGER.info("codeString " + codeString);

            newLine = System.getProperty("line.separator");
            seedTopFileLineSeparator = SystemUtils.getCfg(GlobalConstantDef.CONFIG_PATH).getProperty(
                    "top100_file_line_separator");
        }
        catch (Exception e)
        {
            LOGGER.error("读取Top100文件错误,文件未找到", e);
            return null;
        }

        File file = seedsFile;
        FileOutputStream fileOutStream = null;
        OutputStreamWriter outSW = null;
        BufferedWriter bw = null;

        try
        {
            fileOutStream = new FileOutputStream(file);
            outSW = new OutputStreamWriter(fileOutStream, codeString);
            bw = new BufferedWriter(outSW);
            // 分割top文件,生成小文件
            for (int j = 0; j < splitFileLine; j++)
            {
                String readLine;
                readLine = br.readLine();
                if (StringUtils.isNotBlank(readLine))
                {
                    String[] split = readLine.split(seedTopFileLineSeparator);
                    StringBuffer sBuffer = new StringBuffer();
                    for (String str : split)
                    {
                        sBuffer.append(str + seedTopFileLineSeparator);
                    }
                    bw.write(sBuffer + newLine);
                }
            }
            // 最后一个节点分配余下所有的域名
            if (loopNum == (nodeSize - 1))
            {
                String tempLine = null;
                while ((tempLine = br.readLine()) != null)
                {
                    bw.write(tempLine + newLine);
                }
            }
            bw.flush();
        }
        catch (IOException e)
        {
            LOGGER.error("当下发爬虫任务,拆分Top文件时失败", e);
            return null;
        }
        finally
        {
            IOUtils.closeStream(br);
            IOUtils.closeStream(bw);
            IOUtils.closeStream(outSW);
            IOUtils.closeStream(fileOutStream);
        }
        return file;
    }

    // 处理Dns解析任务下发
    public static List<IDCNode> handleDnsParseTaskUnder(IMSPClient client, List<IDCNode> nodeList, String taskName,
            String taskBeginTime, File underFile) throws IOException
    {
        int resultcode = -1;
        int nodeSize = nodeList.size();
        List<IDCNode> listForFailNode = new ArrayList<IDCNode>();

        // 循环所有可用节点,下发所有任务
        for (int i = 0; i < nodeSize; i++)
        {
            IDCNode nodeInfo = nodeList.get(i);
            // String mnidStr = nodeInfo.getMnId();
            // int mnId = Integer.valueOf(mnidStr);

            // 下发任务
            resultcode = CommonUnderTask.createTask(client, taskName, "dns_parse_tool_type", underFile, nodeInfo,
                    taskBeginTime);

            // 失败节点信息
            if (resultcode != 0)
            {
                listForFailNode.add(nodeInfo);
            }
        }
        return listForFailNode;
    }

    private static BufferedReader readTopFile() throws IOException
    {
        LOGGER.info("readTopFile IN....");

        String seedTopFileName = SystemUtils.getCfg(GlobalConstantDef.CONFIG_PATH).getProperty("top100_file_name");
        LOGGER.info("Top 文件名称.... " + seedTopFileName);

        BufferedReader br = null;

        br = new BufferedReader(new FileReader(SystemUtils.class.getResource("/").getFile() + seedTopFileName));

        LOGGER.info("readTopFile out....");
        return br;
    }

}

class NodeUseInfo
{
    public int getMnCode()
    {
        return mnCode;
    }

    public void setMnCode(int mnCode)
    {
        this.mnCode = mnCode;
    }

    public boolean isAvailableFlag()
    {
        return availableFlag;
    }

    public void setAvailableFlag(boolean availableFlag)
    {
        this.availableFlag = availableFlag;
    }

    public boolean isUseFlag()
    {
        return useFlag;
    }

    public void setUseFlag(boolean useFlag)
    {
        this.useFlag = useFlag;
    }

    private int mnCode;
    private boolean availableFlag;
    private boolean useFlag;
    private IDCNode node;

    /**
     * @return node
     */
    public IDCNode getNode()
    {
        return node;
    }

    /**
     * @param node
     *            要设置的 node
     */
    public void setNode(IDCNode node)
    {
        this.node = node;
    }

}
